package io.kwdb.examples;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollDomainSocketChannel;
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.unix.DomainSocketAddress;
import io.kwdb.sdk.Batcher;
import io.kwdb.sdk.Config;
import io.kwdb.sdk.Point;
import io.kwdb.sdk.Status;
import io.kwdb.sdk.StreamingBatchClient;
import io.kwdb.sdk.internal.Metrics;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/** Example demonstrating Phase-2 streaming with batching, retry, and metrics. */
public class StreamingExample {

  public static void main(String[] args) throws Exception {
    System.out.println("=== Java Streaming Batch Client Example (Phase-2) ===\n");

    // Parse command line arguments
    int totalRecords = 100000; // Default
    if (args.length > 0) {
      try {
        totalRecords = Integer.parseInt(args[0]);
      } catch (NumberFormatException e) {
        System.err.println("Invalid record count, using default: " + totalRecords);
      }
    }

    // Test both TCP and UDS connections
    testTcpBatchStreaming(totalRecords);
    testUdsBatchStreaming(totalRecords);

    System.out.println("\n=== All streaming tests completed ===");
  }

  private static void testTcpBatchStreaming(int totalRecords) throws Exception {
    System.out.println("[Test 1] Testing TCP batch streaming with Phase-2 features...");
    System.out.println(
        "Configuration: " + totalRecords + " records, 1000 batch size, 100ms linger");

    // Create channel
    ManagedChannel channel =
        ManagedChannelBuilder.forAddress("127.0.0.1", 9090).usePlaintext().build();

    try {
      // Create client with Phase-2 configuration
      StreamingBatchClient client =
          StreamingBatchClient.builder(channel)
              .maxBatchSize(1000)
              .lingerMs(100)
              .maxQueueSize(10000)
              .backpressureStrategy(Config.Backpressure.BLOCK)
              .enableRetry(true)
              .maxRetries(3)
              .baseBackoffMs(100)
              .build();

      // Start client
      CompletableFuture<Status> startFuture = client.start();
      Status startStatus = startFuture.get(5, TimeUnit.SECONDS);

      if (!startStatus.isOk()) {
        System.err.println("Failed to start client: " + startStatus.getMessage());
        return;
      }

      System.out.println("Stream started successfully");

      // Write points
      long startTime = System.nanoTime();
      int successCount = 0;
      int failureCount = 0;

      for (int i = 0; i < totalRecords; i++) {
        Point point =
            Point.builder("java_stream_perf")
                .tag("host", "java-host-" + (i % 10))
                .tag("region", "us-" + (i % 3))
                .field("value", i * 1.5)
                .field("temperature", 20.0 + Math.random() * 10)
                .timestampNow()
                .build();

        Status status = client.write(point);
        if (status.isOk()) {
          successCount++;
        } else {
          failureCount++;
        }

        // Progress indicator every 10k points
        if ((i + 1) % 10000 == 0) {
          System.out.println("Progress: " + (i + 1) + "/" + totalRecords + " points");
        }
      }

      System.out.println("All points submitted. Flushing...");

      // Flush and close
      client.flush();
      Status closeStatus = client.close(30, TimeUnit.SECONDS);

      long elapsedNs = System.nanoTime() - startTime;
      double elapsedSec = elapsedNs / 1_000_000_000.0;
      double qps = totalRecords / elapsedSec;

      // Print metrics
      printMetrics("TCP", client, successCount, failureCount, elapsedSec, qps);

      if (!closeStatus.isOk()) {
        System.err.println("Warning: Close status: " + closeStatus.getMessage());
      }

    } finally {
      channel.shutdown();
      channel.awaitTermination(5, TimeUnit.SECONDS);
    }
  }

  private static void testUdsBatchStreaming(int totalRecords) throws Exception {
    System.out.println("\n[Test 2] Testing UDS batch streaming with Phase-2 features...");

    // Check if UDS is available (Linux only)
    String os = System.getProperty("os.name").toLowerCase();
    if (!os.contains("linux")) {
      System.out.println("UDS not available on " + os + ", skipping test");
      return;
    }

    System.out.println(
        "Configuration: " + totalRecords + " records, 1000 batch size, 100ms linger");

    // Create UDS channel
    ManagedChannel channel =
        NettyChannelBuilder.forAddress(new DomainSocketAddress("/tmp/kwdb-ingestd.sock"))
            .channelType(EpollDomainSocketChannel.class)
            .eventLoopGroup(new EpollEventLoopGroup())
            .usePlaintext()
            .build();

    try {
      // Create client with Phase-2 configuration
      StreamingBatchClient client =
          StreamingBatchClient.builder(channel)
              .maxBatchSize(1000)
              .lingerMs(100)
              .maxQueueSize(10000)
              .backpressureStrategy(Config.Backpressure.BLOCK)
              .enableRetry(true)
              .maxRetries(3)
              .baseBackoffMs(100)
              .build();

      // Start client
      CompletableFuture<Status> startFuture = client.start();
      Status startStatus = startFuture.get(5, TimeUnit.SECONDS);

      if (!startStatus.isOk()) {
        System.err.println("Failed to start UDS client: " + startStatus.getMessage());
        return;
      }

      System.out.println("UDS stream started successfully");

      // Write points
      long startTime = System.nanoTime();
      int successCount = 0;
      int failureCount = 0;

      for (int i = 0; i < totalRecords; i++) {
        Point point =
            Point.builder("java_uds_perf")
                .tag("host", "uds-host-" + (i % 10))
                .tag("app", "streaming-test")
                .field("latency", Math.random() * 100)
                .field("throughput", Math.random() * 1000)
                .timestampNow()
                .build();

        Status status = client.write(point);
        if (status.isOk()) {
          successCount++;
        } else {
          failureCount++;
        }

        // Progress indicator every 10k points
        if ((i + 1) % 10000 == 0) {
          System.out.println("Progress: " + (i + 1) + "/" + totalRecords + " points");
        }
      }

      System.out.println("All points submitted. Flushing...");

      // Flush and close
      client.flush();
      Status closeStatus = client.close(30, TimeUnit.SECONDS);

      long elapsedNs = System.nanoTime() - startTime;
      double elapsedSec = elapsedNs / 1_000_000_000.0;
      double qps = totalRecords / elapsedSec;

      // Print metrics
      printMetrics("UDS", client, successCount, failureCount, elapsedSec, qps);

      if (!closeStatus.isOk()) {
        System.err.println("Warning: Close status: " + closeStatus.getMessage());
      }

    } finally {
      channel.shutdown();
      channel.awaitTermination(5, TimeUnit.SECONDS);
    }
  }

  /** Print detailed metrics including P50/P95/P99 latency and QPS. */
  private static void printMetrics(
      String label,
      StreamingBatchClient client,
      int successCount,
      int failureCount,
      double elapsedSec,
      double qps) {
    System.out.println("\n=== " + label + " Performance Metrics ===");

    // Basic stats
    System.out.println("Total Time: " + String.format("%.2f", elapsedSec) + " seconds");
    System.out.println("QPS: " + String.format("%.0f", qps) + " points/s");
    System.out.println("Writes OK: " + successCount);
    System.out.println("Writes Failed: " + failureCount);

    double successRate = (successCount * 100.0) / (successCount + failureCount);
    System.out.println("Success Rate: " + String.format("%.2f%%", successRate));

    // Client metrics
    System.out.println("Total Points Sent: " + client.getTotalPointsSent());
    System.out.println("Total Rows Written: " + client.getTotalRowsWritten());

    // Batcher stats
    Batcher.BatcherStats batcherStats = client.getBatcherStats();
    System.out.println("Batches Submitted: " + batcherStats.submittedBatches);
    System.out.println("Items Submitted: " + batcherStats.submittedItems);
    System.out.println("Items Dropped: " + batcherStats.droppedItems);
    System.out.println("Queue Size: " + batcherStats.currentQueueSize);

    // Detailed metrics from Metrics component
    Metrics.MetricsSnapshot metricsSnapshot = client.getMetrics();
    System.out.println("\n--- Latency Statistics ---");
    System.out.println("Batch Latency: " + metricsSnapshot.batchLatency);
    System.out.println("Write Latency: " + metricsSnapshot.writeLatency);

    if (!metricsSnapshot.writesFailed.isEmpty()) {
      System.out.println("\nFailed Writes by Error Code: " + metricsSnapshot.writesFailed);
    }

    System.out.println("=========================================\n");
  }
}
